/*================================================================
*   Copyright (C) 2020 Sangfor Ltd. All rights reserved.
*   
*   文件名称：ThreadPool.cpp
*   创 建 者：Ann
*   创建日期：2020年06月13日
*   描    述：
*
================================================================*/

#include <unistd.h>
#include <pthread.h>
#include <iostream>
#include <queue>

#define THREADCOUNT 4
#define QUEUESIZE 1

typedef void (*Handler_t)(int);

class ThreadTask
{
    public:
        ThreadTask(int data, Handler_t handler)
        {
            data_ = data; // 0 - 99
            handler_ = handler; // func
        }

        void Run()
        {
            //func(0~99)
            handler_(data_);
        }
    private:
        //待处理的数据
        int data_;
        //保存处理函数地址的函数指针
        Handler_t handler_;
};

class ThreadPool
{
    public:
        ThreadPool()
        {
            Capacity_ = QUEUESIZE;

            pthread_mutex_init(&lock_, NULL);
            pthread_cond_init(&cond_, NULL);

            bool is_create = CreateThreads();
            if(!is_create)
            {
                printf("ThreadPool Create failed\n");
                exit(1);
            }

            is_quit_ = false;
        }

        ~ThreadPool()
        {
            pthread_mutex_destroy(&lock_);
            pthread_cond_destroy(&cond_);
        }

        bool Push(ThreadTask* tt)
        {
            pthread_mutex_lock(&lock_);
            if(is_quit_)
            {
                pthread_mutex_unlock(&lock_);
                return false;
            }
            que_.push(tt);
            pthread_mutex_unlock(&lock_);

            //通知线程池当中的线程来进行处理
            pthread_cond_signal(&cond_);
            return true;
        }

        void Pop(ThreadTask** tt)
        {
            *tt = que_.front();
            que_.pop();
        }

        void ThreadPoolClear()
        {
            pthread_mutex_lock(&lock_);
            is_quit_ = true;
            pthread_mutex_unlock(&lock_);

            pthread_cond_broadcast(&cond_);
        }

        void ThreadJoin()
        {
            for(int i = 0; i < THREADCOUNT; i++)
            {
                pthread_join(tid_[i], NULL);
            }
        }
    private:
        static void* Thread_start(void* arg)
        {
            //其实就是强转为线程池的实例化指针
            ThreadPool* tp = (ThreadPool*)arg;
            while(1)
            {
                //加锁
                pthread_mutex_lock(&tp->lock_);
                while(tp->que_.empty())
                {
                    if(tp->is_quit_)
                    {
                        //kill当前进程
                        pthread_mutex_unlock(&tp->lock_);
                        pthread_exit(NULL);
                    }
                    pthread_cond_wait(&tp->cond_, &tp->lock_);
                }
                ThreadTask* tt;
                tp->Pop(&tt);
                pthread_mutex_unlock(&tp->lock_);

                tt->Run();
                delete tt;
            }
        }

        bool CreateThreads()
        {
            for(int i = 0; i < THREADCOUNT; i++)
            {
                int ret = pthread_create(&tid_[i], NULL, Thread_start, (void*)this);
                if(ret < 0)
                {
                    perror("pthread_create");
                    return false;
                }
            }

            return true;
        }
    private:
        std::queue<ThreadTask*> que_;
        size_t Capacity_;

        //互斥
        pthread_mutex_t lock_;

        //同步 --> 只定义消费的条件变量， 因为请求什么时候来，什么时候入队操作，我们后台是无法把控
        pthread_cond_t cond_;

        //保存了线程池当中的线程标识符
        pthread_t tid_[THREADCOUNT];

        bool is_quit_;
};


void func(int data)
{
    printf("data = [%p][%d]\n", pthread_self(), data);
}

int main()
{
    ThreadPool* tp = new ThreadPool();

    //起到一个生产者的作用，往线程池的队列当中去push数据
    for(int i = 0; i < 100; i++)
    {
        ThreadTask* tt = new ThreadTask(i, func);
        tp->Push(tt);
    }

    sleep(10);
    tp->ThreadPoolClear();
    tp->ThreadJoin();

    delete tp;
    return 0;
}
